///*
// * Copyright 2012 The Netty Project
// *
// * The Netty Project licenses this file to you under the Apache License,
// * version 2.0 (the "License"); you may not use this file except in compliance
// * with the License. You may obtain a copy of the License at:
// *
// *   http://www.apache.org/licenses/LICENSE-2.0
// *
// * Unless required by applicable law or agreed to in writing, software
// * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// * License for the specific language governing permissions and limitations
// * under the License.
// */
//package com.lk.service;
//
//import java.nio.charset.StandardCharsets;
//import java.util.HashMap;
//import java.util.LinkedHashMap;
//import java.util.Map;
//import java.util.Objects;
//import java.util.concurrent.ConcurrentHashMap;
//
//import com.ghca.dbproxy.clean.BaseCleanService;
//import com.ghca.dbproxy.common.Constants;
//import com.ghca.dbproxy.exception.BaseException;
//import com.ghca.dbproxy.init.AppConstant;
//import com.ghca.dbproxy.parsing.util.PolicyUtil;
//import com.ghca.dbproxy.server.MyClientSocket;
//import com.ghca.dbproxy.util.ClientUtil;
//import com.ghca.dbproxy.util.PerformanceMonitorUtil;
//import com.ghca.dbproxy.util.i18.I18nUtil;
//import io.netty.handler.logging.LogLevel;
//import io.netty.handler.logging.LoggingHandler;
//import lombok.extern.slf4j.Slf4j;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.http.HttpRequest;
//
//import com.ghca.dbproxy.parsing.ParsingService;
//import com.ghca.dbproxy.parsing.ParsingServiceFactory;
//import com.ghca.dbproxy.server.DBProxySession;
//
//import io.netty.bootstrap.Bootstrap;
//import io.netty.buffer.ByteBuf;
//import io.netty.buffer.Unpooled;
//import io.netty.channel.Channel;
//import io.netty.channel.ChannelFuture;
//import io.netty.channel.ChannelFutureListener;
//import io.netty.channel.ChannelHandlerContext;
//import io.netty.channel.ChannelInboundHandlerAdapter;
//import io.netty.channel.ChannelInitializer;
//import io.netty.channel.ChannelOption;
//import io.netty.channel.EventLoopGroup;
//import io.netty.channel.nio.NioEventLoopGroup;
//import io.netty.channel.socket.SocketChannel;
//import io.netty.channel.socket.nio.NioSocketChannel;
//import io.netty.handler.codec.http.HttpContent;
//@Slf4j
//public class DBProxyFrontendHandler extends ChannelInboundHandlerAdapter {
//
// private static final Logger logger = LoggerFactory.getLogger(DBProxyFrontendHandler.class);
// private static final String CLINET_IP = "CLINET_IP";
// private static final String REDIRECT_IP = "REDIRECT_IP";
// private static final String REDIRECT_PORT = "REDIRECT_PORT";
// private static final String TIME_COMMING = "time_comming";
// public static final String PARM_CONTEXT = "parmContext";
//
// private final int proxyPort;
// private final String remoteHost;
// private final int remotePort;
// private final String remoteType;
// private ResponseMsg responseMsg;
// private Map <String, Object> parm;
// private DBProxyBackendHandler dbProxyBackendHandler;
//
// /**
//  *判断之前是否进行重定向
//  */
// private static Map<String,String> ipMacMap = new HashMap<>();
// // As we use inboundChannel.eventLoop() when building the Bootstrap this does
// // not need to be volatile as
// // the outboundChannel will use the same EventLoop (and therefore Thread) as the
// // inboundChannel.
// private Channel outboundChannel;
//
// public DBProxyFrontendHandler(int proxyPort,String remoteHost, int remotePort, String remoteType, ResponseMsg responseMsg) {
//  this.proxyPort = proxyPort;
//  this.remoteHost = remoteHost;
//  this.remotePort = remotePort;
//  this.remoteType = remoteType;
//  this.responseMsg = responseMsg;
// }
//
// public DBProxyFrontendHandler(int proxyPort,String remoteHost, int remotePort, String remoteType, ResponseMsg responseMsg, Map<String, Object> parm) {
//  this.proxyPort = proxyPort;
//  this.remoteHost = remoteHost;
//  this.remotePort = remotePort;
//  this.remoteType = remoteType;
//  this.responseMsg = responseMsg;
//  this.parm = parm;
// }
//
// @Override
// public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
//
//  super.channelRegistered(ctx);
//
// }
//
// @Override
// public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
//
//  super.channelUnregistered(ctx);
// }
//
// @Override
// public void channelActive(ChannelHandlerContext ctx) {
//  String host = remoteHost;
//  int port = remotePort;
//  final Channel inboundChannel = ctx.channel();
//  String clinetId = ctx.channel().id().asLongText();
//  Map<String,Long> timeMap = new LinkedHashMap<>();
//  timeMap.put("begin",ClientUtil.getCurrentTimeMillis());
//  DBProxySession.getInstance().put(clinetId, TIME_COMMING,timeMap);
//
//  //全链路性能监控，开启socket客户端
//  if(PerformanceMonitorUtil.enable){
//   try{
//    MyClientSocket.createSocket(PerformanceMonitorUtil.gatewayIp,Integer.parseInt(PerformanceMonitorUtil.recordPerformanceMonitorSocketPort));
//   }catch(Exception e){
//    e.printStackTrace();
//   }
//  }
//
//  // Start the connection attempt.
//  Bootstrap b = new Bootstrap();
//  this.dbProxyBackendHandler = new DBProxyBackendHandler(inboundChannel, remoteType, clinetId, responseMsg);
//  b.group(inboundChannel.eventLoop()).channel(ctx.channel().getClass())
//          .handler(dbProxyBackendHandler)
//          .option(ChannelOption.AUTO_READ, false);
//
//  // 激活的时候开始判断是否需要重定向
//  ParsingService parsingService = ParsingServiceFactory.create(remoteType);
//  Map<String, Object> parm = new HashMap<>();
//
//  if (this.parm != null) {
//   this.parm.forEach((k, v) -> parm.merge(k, v, (v1, v2) -> v2));
//  }
//
//  SocketChannel ch = (SocketChannel) ctx.channel();
//  String clinetIP = (String) DBProxySession.getInstance().get(clinetId, CLINET_IP);
//  if (null == clinetIP || "".equals(clinetIP)) {
//   clinetIP = ch.remoteAddress().getAddress().getHostAddress();
//   parm.put(CLINET_IP, clinetIP);
//   DBProxySession.getInstance().put(clinetId, CLINET_IP, clinetIP);
//  }
//  String language = (String)DBProxySession.getInstance().get(clinetId, "CLIENT_LANGUAGE");
//  if (null == language || "".equals(language)) {
//   //默认中文
//   DBProxySession.getInstance().put(clinetId, "CLIENT_LANGUAGE", "zh_CN");
//  }
//
//  byte[] bytes = {};
//  parsingService.doParsing(bytes, clinetId, parm);
//  String isredirect = (String) parm.get("redirect");
//  if(isredirect != null && "2".equals(isredirect)){
//   String redirectIp = (String) parm.get(DBProxyFrontendHandler.REDIRECT_IP);
//   int redirectPort = (int) parm.get(DBProxyFrontendHandler.REDIRECT_PORT);
//   host = redirectIp;
//   port = redirectPort;
//  }
//
//  ChannelFuture f = b.connect(host, port);
//  DBProxySession.getInstance().put(clinetId, "cf", f);
//  outboundChannel = f.channel();
//  //后端流量日志打印
//  if (ClientUtil.isDownflowLogStatic) {
//   outboundChannel.pipeline().addFirst(new LoggingHandler(LogLevel.INFO));
//  }
//  f.addListener(new ChannelFutureListener() {
//   @Override
//   public void operationComplete(ChannelFuture future) {
//    if (future.isSuccess()) {
//     logger.info("con target server succ >>[remoteHost]>> " + remoteHost + " >> [remotePort]>>"
//             + remotePort);
//     inboundChannel.read();
//    } else {
//     logger.info(
//             "con target server fd >>[remoteHost]>> " + remoteHost + " >> [remotePort]" + remotePort);
//     inboundChannel.close();
//    }
//   }
//  });
// }
//
// @Override
// public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
//  String clientId = ctx.channel().id().asLongText();
//  LinkedHashMap<Object, Object> timeMap = (LinkedHashMap<Object, Object>) DBProxySession.getInstance().get(clientId, TIME_COMMING);
//  if ( timeMap == null){
//   timeMap = new LinkedHashMap<>();
//   DBProxySession.getInstance().put(clientId, TIME_COMMING,timeMap);
//  }
//  timeMap.put("begin",ClientUtil.getCurrentTimeMillis());
//
//  if (msg instanceof ByteBuf) {
//   ByteBuf buffer = (ByteBuf) msg;
//   byte[] bytes = new byte[buffer.readableBytes()];
//   buffer.readBytes(bytes);
//   buffer.release();
//   SocketChannel ch = (SocketChannel) ctx.channel();
//
//   timeMap.put("releasebuffer", ClientUtil.getCurrentTimeMillis());
//   ParsingService parsingService = ParsingServiceFactory.create(remoteType);
//   timeMap.put("create", ClientUtil.getCurrentTimeMillis());
//
//   Map<String, Object> parm = (Map<String, Object>) DBProxySession.getInstance().get(clientId, PARM_CONTEXT);
//   if (parm == null) {
//    parm = new HashMap<>();
//    buildDBProxySession(parm, clientId, ch);
//    DBProxySession.getInstance().put(clientId, PARM_CONTEXT, parm);
//   }
//
//   timeMap.put("buildDBProxySession",ClientUtil.getCurrentTimeMillis());
//
//   byte[] newmsg = parsingService.doParsing(bytes, clientId, parm);
//
//   String isredirect = (String) parm.get("redirect");
//
//   //mongodb 关闭连接报文
//   String mongodbClose = "MongoDB close session";
//   if(null != newmsg) {
//    String mongodbCloseByte = new String(newmsg, StandardCharsets.UTF_8);
//    if (Objects.equals(mongodbCloseByte, mongodbClose)) {
//     channelUnregistered(ctx);
//     return;
//    }
//   }
//
//   //hbase关闭连接
//   String closeConn = (String) DBProxySession.getInstance().get(clientId, "closeConn");
//   if (closeConn != null){
//    DBProxySession.getInstance().put(clientId, "closeConn", null);
//    // 客户端提示Connection closed。
//    ctx.channel().close();
//    //直接给客户端发送异常报文。
//    return;
//   }
//
//   if (null == newmsg) {
//    newmsg = new byte[0];
//    ByteBuf buf = Unpooled.wrappedBuffer(newmsg);
//
//    Channel inboundChannel = ctx.channel();
//    inboundChannel.writeAndFlush(buf).addListener(new ChannelFutureListener() {
//     @Override
//     public void operationComplete(ChannelFuture future) {
//      if (future.isSuccess()) {
//       ctx.channel().read();
//      } else {
//       future.channel().close();
//      }
//     }
//    });
//   } else if ("1".equalsIgnoreCase(isredirect)) {
//    isRedirect1(ctx,newmsg);
//   } else if ("2".equalsIgnoreCase(isredirect)) {
//    isRedirect2(ctx,newmsg,parm,clientId);
//   } else if ("3".equalsIgnoreCase(isredirect)) {
//    isRedirect3(ctx,newmsg,parm,clientId);
//   } else if (outboundChannel.isActive()) {
//    isActive(ctx,newmsg);
//   } else {
//    //之前就没有
//   }
//  }
// }
//
// private void buildDBProxySession(Map<String,Object> parm,String clientId,SocketChannel ch){
//  parm.put("remote_ip",remoteHost);
//  parm.put("remote_port",remotePort);
//  parm.put("proxy_port",proxyPort);
//  String clientIp = (String) DBProxySession.getInstance().get(clientId, CLINET_IP);
//  if (null == clientIp || "".equals(clientIp)) {
//   clientIp = ch.remoteAddress().getAddress().getHostAddress();
//   parm.put(CLINET_IP, clientIp);
//   DBProxySession.getInstance().put(clientId, CLINET_IP, clientIp);
//  }
//  String clinetMac = "";
//  if(ipMacMap.containsKey(clientIp)){
//   clinetMac = ipMacMap.get(clientIp);
//  }
//  parm.put("CLINET_MAC", clinetMac);
//  DBProxySession.getInstance().put(clientId, "CLINET_MAC", clinetMac);
//  DBProxySession.getInstance().put(clientIp, "parm", parm);
//
//  //运维平台迁移代码 缓存源端口号后续访问PAM获取用户真实信息使用---------------------------------------------
//  DBProxySession.getInstance().put(clientId,"CLINET_SOURCE_PORT",ch.remoteAddress().getPort());
//  //新增CLINET_SOURCE_IP
//  DBProxySession.getInstance().put(clientId,"CLINET_SOURCE_IP",ch.remoteAddress().getAddress().getHostAddress());
//
//  if (this.parm != null) {
//   this.parm.forEach((k, v) -> parm.merge(k, v, (v1, v2) -> v2));
//  }
// }
//
// @Override
// public void channelInactive(ChannelHandlerContext ctx) {
//  if (outboundChannel != null) {
//   closeOnFlush(outboundChannel);
//  }
// }
//
//
//
// private void isActive(ChannelHandlerContext ctx,byte[] newmsg){
//  ByteBuf buf = Unpooled.wrappedBuffer(newmsg);
//  /**
//   * 全链路性能监控
//   * 节点：向数据库发送请求
//   */
//  if(PerformanceMonitorUtil.enable){
//   String clintID = ctx.channel().id().asLongText();
//   if(DBProxySession.getInstance().get(clintID,"PERFORMANCE_MONITOR_ISSYSTEMSQL") != null && !(boolean)DBProxySession.getInstance().get(clintID,"PERFORMANCE_MONITOR_ISSYSTEMSQL")){
//    //不是系统sql
//    PerformanceMonitorUtil.performanceMonitorRecord_reqForDB(clintID,PerformanceMonitorUtil.now());
//   }
//  }
//  String clintID = ctx.channel().id().asLongText();
//  ((Map)DBProxySession.getInstance().get(clintID, TIME_COMMING)).put("upflowBeforeSend",ClientUtil.getCurrentTimeMillis());
//  outboundChannel.writeAndFlush(buf).addListener(new ChannelFutureListener() {
//   @Override
//   public void operationComplete(ChannelFuture future) {
//    ((Map)DBProxySession.getInstance().get(clintID, TIME_COMMING)).put("upflowSendOk",ClientUtil.getCurrentTimeMillis());
//    if (future.isSuccess()) {
//     // was able to flush out data, start to read the next chunk
//     ctx.channel().read();
//    } else {
//     future.channel().close();
//    }
//   }
//  });
// }
//
// private void isRedirect1(ChannelHandlerContext ctx,byte[] newmsg){
//  ByteBuf buf = Unpooled.wrappedBuffer(newmsg);
//  Channel inboundChannel = ctx.channel();
//  inboundChannel.writeAndFlush(buf).addListener(new ChannelFutureListener() {
//   @Override
//   public void operationComplete(ChannelFuture future) {
//    if (future.isSuccess()) {
//     ctx.channel().read();
//    } else {
//     future.channel().close();
//    }
//   }
//  });
// }
// private void isRedirect2(ChannelHandlerContext ctx,byte[] newmsg,Map<String,Object> parm,String clientId) throws BaseException {
//
//  try {
//   // 增加部分功能可以登录完成后进行重定向，目前注释中
//   String redirectIp = (String) parm.get(DBProxyFrontendHandler.REDIRECT_IP);
//   int redirectPort = (int) parm.get(DBProxyFrontendHandler.REDIRECT_PORT);
//   ByteBuf newmsgBuf = Unpooled.wrappedBuffer(newmsg);
//   ChannelFuture cfs = (ChannelFuture) DBProxySession.getInstance().get(clientId, "cf");
//   if (cfs != null && cfs.channel().isActive()/* && redirectFlag*/) {
//    cfs.channel().writeAndFlush(newmsgBuf).addListener(new ChannelFutureListener() {
//     @Override
//     public void operationComplete(ChannelFuture future) {
//      if (future.isSuccess()) {
//       // was able to flush out data, start to read the next chunk
//       ctx.channel().read();
//      } else {
//       future.channel().close();
//      }
//     }
//    });
//   } else {
//    Bootstrap b = new Bootstrap();
//    Channel inboundChannel = ctx.channel();
//    b.group(inboundChannel.eventLoop()).channel(inboundChannel.getClass())
//            .handler(new ChannelInboundHandlerAdapter() {
//             @Override
//             public void channelActive(ChannelHandlerContext ctx) {
//              ctx.read();
//             }
//
//             @Override
//             public void channelRead(final ChannelHandlerContext ctx, Object msg) {
//
//              inboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
//               @Override
//               public void operationComplete(ChannelFuture future) {
//                if (future.isSuccess()) {
//                 ctx.channel().read();
//                } else {
//                 future.channel().close();
//                }
//               }
//              });
//             }
//
//             @Override
//             public void channelInactive(ChannelHandlerContext ctx) {
//              closeOnFlush(inboundChannel);
//             }
//
//             @Override
//             public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
//              cause.printStackTrace();
//              closeOnFlush(ctx.channel());
//             }
//            }).option(ChannelOption.AUTO_READ, false);
//
//    ChannelFuture cf = b.connect(redirectIp, redirectPort);
//    DBProxySession.getInstance().put(clientId, "cf", cf);
//    cf.addListener(new ChannelFutureListener() {
//     @Override
//     public void operationComplete(ChannelFuture future) {
//      if (future.isSuccess()) {
//       logger.info("con target server succ >>[remoteHost]>> " + remoteHost
//               + " >> [remotePort]>>" + remotePort);
//       inboundChannel.read();
//       if (cf.channel().isActive()) {
//        cf.channel().writeAndFlush(newmsgBuf).addListener(new ChannelFutureListener() {
//         @Override
//         public void operationComplete(ChannelFuture future) {
//          if (future.isSuccess()) {
//           // was able to flush out data, start to read the next chunk
//           ctx.channel().read();
//          } else {
//           future.channel().close();
//          }
//         }
//        });
//       }
//
//      } else {
//       logger.info("con target server fd >>[remoteHost]>> " + remoteHost + " >> [remotePort]"
//               + remotePort);
//       future.channel().close();
//      }
//     }
//    });
//   }
//  } catch (Exception e) {
//   throw new BaseException(e.toString());
//  }
// }
//
//
// private void isRedirect3(ChannelHandlerContext ctx,byte[] newmsg,Map<String,Object> parm,String clientId){
//
//  try {
//   String redirectIp = (String) parm.get(DBProxyFrontendHandler.REDIRECT_IP);
//   int redirectPort = (int) parm.get(DBProxyFrontendHandler.REDIRECT_PORT);
//   Channel inboundChannel = ctx.channel();
//   EventLoopGroup group = new NioEventLoopGroup();
//   Bootstrap b = new Bootstrap();
//   b.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
//    @Override
//    protected void initChannel(SocketChannel sc) throws Exception {
//     sc.pipeline().addLast(new ChannelInboundHandlerAdapter() {
//      @Override
//      public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//       try {
//        // 回写重定向数据库的响应消息
//        ByteBuf buffer = (ByteBuf) msg;
//
//        inboundChannel.writeAndFlush(Unpooled.copiedBuffer(buffer))
//                .addListener(new ChannelFutureListener() {
//                 @Override
//                 public void operationComplete(ChannelFuture future) {
//                  if (future.isSuccess()) {
//                   inboundChannel.read();
//                   //
//                   group.shutdownGracefully();
//                  } else {
//                   future.channel().close();
//                  }
//                 }
//                });
//       } finally {
//        // 但只读数据时,!必须手动释放引用数
//       }
//      }
//
//     });
//    }
//   });
//
//   ChannelFuture cf1 = b.connect(redirectIp, redirectPort).sync();
//   cf1.addListener(new ChannelFutureListener() {
//    @Override
//    public void operationComplete(ChannelFuture arg0) throws Exception {
//     if (cf1.isSuccess()) {
//
//      log.info(I18nUtil.getDefaultLangMsg(Constants.LOG_INFO_000117));
//     } else {
//      log.info(I18nUtil.getDefaultLangMsg(Constants.LOG_INFO_000118));
//      logger.info(I18nUtil.getDefaultLangMsg(Constants.LOG_INFO_000118));
//      cf1.cause().printStackTrace();
//      group.shutdownGracefully(); // 关闭线程组
//     }
//    }
//   });
//
//   // 发送消息, Buffer类型. write需要flush才发送, 可用writeFlush代替
//   ByteBuf newmsgBuf = Unpooled.wrappedBuffer(newmsg);
//   cf1.channel().writeAndFlush(Unpooled.copiedBuffer(newmsgBuf));
//
//  } catch (InterruptedException e) {
//   logger.error(e.getMessage());
//   Thread.currentThread().interrupt();
//  }
// }
//
// @Override
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
//  logger.info("exceptionCaught, " , cause);
//  cause.printStackTrace();
//  closeOnFlush(ctx.channel());
// }
//
// /**
//  * Closes the specified channel after all queued write requests are flushed.
//  */
// static void closeOnFlush(Channel ch) {
//  if (ch.isActive()) {
//   ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
//  }
// }
//
// @Override
// public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
//
//  super.handlerAdded(ctx);
//  // 創建session
//  String clinetId = ctx.channel().id().asLongText();
//  DBProxySession.getInstance().create(clinetId);
// }
//
// @Override
// public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
//
//  super.handlerRemoved(ctx);
//  // 銷毀session
//  String clinetId = ctx.channel().id().asLongText();
//  BaseCleanService.sendDownFlowLog(clinetId);
//  DBProxySession.getInstance().removed(clinetId);
// }
//}
